package com.dxf.bigdata.D04_spark

import org.apache.commons.logging.{Log, LogFactory}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 *  RDD方法主要有 转换 和 行动
 *   转换 如  Map , flatMap
 *   行动 collect
 *
 *
 *   RDD算子:   RDD的方法就是算子 --> 对数据的操作称为算子
 *
 *   RDD算子 => 转换算子 Map , 行动算子  collect
 *
 *
 *
 *
 *
 */
object T08_RDD使用_从日志中读取uri {
  private val log: Log = LogFactory.getLog(T08_RDD使用_从日志中读取uri.getClass)
  def main(args: Array[String]): Unit = {

    //TODO 准备环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)

    //TODO 创建RDD

    val rdd: RDD[String] = sc.textFile("datas/apache.log")

    val data: Array[String] = rdd.map(line => {

      val datas: Array[String] = line.split(" ")
      val aa = datas(6)
      aa
    }).collect()

    val rdd2: RDD[String] = sc.makeRDD(data,1)

    rdd2.saveAsTextFile("output")
    //TODO 关闭环境
    sc.stop()


  }
}
